// -*-mode:c++; coding:utf-8-*-

#ifndef _XBASE_ROW_BLOCK_HPP_
#define _XBASE_ROW_BLOCK_HPP_

#include <vector>
#include <algorithm>

#include <boost/intrusive_ptr.hpp>
#include <boost/noncopyable.hpp>
#include <boost/asio/buffer.hpp>
#include <boost/foreach.hpp>

#include <config.hpp>
#include <intrusive_ptr_helper.hpp>
#include <wrapper.hpp>

namespace xbase
{

	namespace detail
	{
		// 管理数据块列表的分配和释放
		struct row_blocks_holder
			: private boost::noncopyable
		{
			typedef default_char_allocator allocator_type;
			typedef boost::fast_pool_allocator<row_blocks_holder> self_allocator_type;
			typedef basic_factory<row_blocks_holder> factory_type;

			allocator_type m_allocator;
			std::vector<boost::asio::const_buffer> m_blocks;
			std::size_t m_bytes;
			reference_counter m_reference_counter;
			INTRUSIVE_PTR_HOOKS(row_blocks_holder, m_reference_counter, factory_type);

			row_blocks_holder()
				: m_bytes(0) {}

			~row_blocks_holder() {
				BOOST_FOREACH(const boost::asio::const_buffer &data, m_blocks)
				{
					m_allocator.deallocate(const_cast<char *>(boost::asio::buffer_cast<const char*>(data)),
							       boost::asio::buffer_size(data));
					m_bytes -= boost::asio::buffer_size(data);
				}
				assert(m_bytes == 0);
			}

			void add_block(const boost::asio::const_buffer &block) {
				m_blocks.push_back(block);
				m_bytes += boost::asio::buffer_size(block);
			}
		};
		
		typedef boost::intrusive_ptr<row_blocks_holder> row_blocks_holder_ptr;

	} // namespace detail

	// 这样实现，空间占用没有增加，也保证了一条kv记录的数据紧挨在一起，便于处理，也便于扫描和读取操作。
	// 但这样一来，一次读操作就要查找两次
	//
	// 行块中的记录不需要是有序的，因为构建索引的时候，会对行块索引
	// 排序。
	//
	// 但要求新写入的数据，在放在最前面；而排序也用稳定排序，从而保
	// 证了新写入的记录在前（这对修改以前存在的记录的操作很重要）。
	class row_block
		: private boost::noncopyable
	{
	public:
		typedef string_wrapper<uint32_t, char, kb::network_byteorder> string_type;
		typedef boost::fast_pool_allocator<row_block> self_allocator_type;
		typedef boost::intrusive_ptr<row_block> pointer;
		
		struct row_block_factory
		{
			typedef row_block::self_allocator_type allocator_type;

			template<typename BufferSequence>
			static inline
			row_block *construct(const BufferSequence &data,
					     const std::size_t bytes) {
				allocator_type m_allocator;
				row_block *p = m_allocator.allocate();
				new(p) row_block(data, bytes);
				return p;
			}

			static inline
			void destroy(row_block *p) {
				allocator_type m_allocator;
				p->~row_block();
				m_allocator.deallocate(p);
			}
		};
		typedef row_block_factory factory_type;

		typedef detail::row_blocks_holder::allocator_type allocator_type;
		typedef detail::row_blocks_holder_ptr row_blocks_holder_ptr;

	private:

		// 行数据的内存块，kv记录就一条一条保存在里面
		// 共享的数据块，每次拷贝时会将原来的数据块保存在这里，以确保它们不被修改。
		row_blocks_holder_ptr m_shared_blocks;

		allocator_type m_allocator;
		boost::asio::const_buffer m_block; // 新的数据将添加到这里
		// 数据只同时存在上面其中一个地方：新的数据块被保存在m_block中，与已有的行块合并时，会放到m_shared_blocks中。
		
		const string_type *m_row_name; // 指向其中一条kv记录中的行名，通常指向第一条记录的行名，即数据的起始位置
		
		block_index_type m_block_index; // 行数据的索引，即指针列表，便于查找；该索引仅在插入到内存表中时才需要建立

		reference_counter m_reference_counter;
		INTRUSIVE_PTR_HOOKS(row_block, m_reference_counter, factory_type);

	public:

		// 传入属于同一行的记录列表
		//
		// 传入的数据要拷贝到自己管理的内存中
		template<typename BufferSequence>
		row_block(const BufferSequence &data,
			  const std::size_t bytes)
			: m_shared_blocks()
			, m_block()
			, m_row_name(NULL)
			, m_block_index() {
			copy_data(data, bytes);
		}

		~row_block() {
			if(boost::asio::buffer_size(m_block) != 0)
			{
				m_allocator.deallocate(const_cast<char *>(boost::asio::buffer_cast<const char*>(m_block)),
						       boost::asio::buffer_size(m_block));
				m_block = boost::asio::const_buffer();
			}
		}

		template<typename BufferSequence>
		std::size_t copy_data(const BufferSequence &data,
				      const std::size_t bytes);

		// 新的行块加入内存表前，要做此操作！
		// 
		// 要求m_shared_blocks必须为空、m_block中必须有一个块
		void share() {
			assert(! m_shared_blocks);
			assert(boost::asio::buffer_size(m_block) != 0);
				
			m_shared_blocks.reset(detail::row_blocks_holder::factory_type::construct());
			m_shared_blocks->add_block(m_block);
			m_block = boost::asio::const_buffer();
		}

		// 调用该函数时，必须保证：
		// 
		// this是一个新创建的行块，只包含一个块数据，并且保存在
		// m_block中；即通过构造函数创建后，没再修改过自身；
		//
		// other是一个新创建的行块（数据只放在m_block中）或已经
		// 包含多个块的行块（数据只放在m_shared_blocks中）；即
		// 只是通过构造函数或merge_from操作过的行块。
		//
		// other被认为是已经存在的、旧的行块数据；this被认为是新的行块数据
		void share_from(const row_block &other) {
			assert(*row_name() == *other.row_name());
			assert(! m_shared_blocks);
			assert(other.m_shared_blocks);

			// share block
			m_shared_blocks = other.m_shared_blocks;
			m_shared_blocks->add_block(m_block);
			m_block = boost::asio::const_buffer(); // must clear, 否则会重复释放
			
			// merge index
			block_index_type new_index(m_block_index.size() + other.m_block_index.size());
			std::merge(m_block_index.begin(), m_block_index.end(),
				   other.m_block_index.begin(), other.m_block_index.end(),
				   new_index.begin(),
				   row_block::entry_less_than());
			m_block_index.swap(new_index);
		}

		const string_type *row_name() const {
			return m_row_name;
		}

		std::size_t data_bytes() const {
			return boost::asio::buffer_size(m_block) +
				((! m_shared_blocks)
				 ? 0
				 : m_shared_blocks->m_bytes);
		}

		const block_index_type &block_index() const {
			return m_block_index;
		}

		bool operator<(const row_block &other) const {
			return (*row_name()) == (*other.row_name());
		}
		bool operator==(const row_block &other) const {
			return (*row_name()) == (*other.row_name());
		}

		bool operator<(const string_type &row) const {
			return (*row_name()) == row;
		}
		bool operator==(const string_type &row) const {
			return (*row_name()) == row;
		}

		bool operator<(const string_type *row) const {
			return (*row_name()) == *row;
		}
		bool operator==(const string_type *row) const {
			return (*row_name()) == *row;
		}

		// 块内的记录比较不再需要比较行了，所以这里实现一个只比较列和时间戳的
		struct entry_less_than
		{
			inline
			bool operator()(const key_wrapper &key1,
					const key_wrapper &key2) const {
				timestamp_less_than timestamp_lt;
				if (*key1.column() == *key2.column())
				{
					return timestamp_lt(*key1.timestamp(), *key2.timestamp());
				}
				else
				{
					return (*key1.column() < *key2.column());
				}
			}
			inline
			bool operator()(const key_wrapper *key1,
					const key_wrapper *key2) const {
				return this->operator()(*key1, *key2);
			}
			inline
			bool operator()(const block_entry &entry1,
					const block_entry &entry2) const {
				return this->operator()(*entry1.m_key, *entry2.m_key);
			}
			inline
			bool operator()(const block_entry *entry1,
					const block_entry *entry2) const {
				return this->operator()(*entry1->m_key, *entry2->m_key);
			}
			inline
			bool operator()(const key_wrapper *key,
					const block_entry *entry) const {
				return this->operator()(*key, *entry->m_key);
			}
			inline
			bool operator()(const block_entry *entry,
					const key_wrapper *key) const {
				return this->operator()(*entry->m_key, *key);
			}
			inline
			bool operator()(const key_wrapper *key,
					const block_entry &entry) const {
				return this->operator()(*key, *entry.m_key);
			}
			inline
			bool operator()(const block_entry &entry,
					const key_wrapper *key) const {
				return this->operator()(*entry.m_key, *key);
			}
		};

		struct entry_equal_to
		{
			inline
			bool operator()(const key_wrapper &key1,
					const key_wrapper &key2) const {
				return (*key1.column() == *key2.column())
					&& *key1.timestamp() == *key2.timestamp();
			}
			inline
			bool operator()(const block_entry &entry1,
					const block_entry &entry2) const {
				return this->operator()(*entry1.m_key, *entry2.m_key);
			}
		};
	};

	template<typename BufferSequence>
	inline
	row_block::pointer make_row_block(const BufferSequence &data,
					  const std::size_t bytes) {
		return row_block::pointer(row_block::factory_type::construct(data, bytes));
	}

	inline
	row_block::pointer make_row_block(const boost::asio::const_buffer &data) {
		return make_row_block(boost::asio::buffer(data),
				      boost::asio::buffer_size(data));
	}

	// copy on write
	// share old_row's data into new_row, and return new_row
	inline
	row_block::pointer share_row_block(row_block::pointer &new_row,
					   row_block::pointer &old_row) {
		new_row->share_from(*old_row);
		return new_row;
	}

	inline
	row_block::pointer share_row_block(row_block::pointer &new_row) {
		new_row->share();
		return new_row;
	}

	
	// 稳定的合并排序，相等的情况下会保证start1在前
	template<typename Compare>
	static inline
	void stable_merge(block_index_type &merged_index,
			  block_index_type::const_iterator start1,
			  block_index_type::const_iterator end1,
			  block_index_type::const_iterator start2,
			  block_index_type::const_iterator end2,
			  const Compare &less)
	{
		while((start1 != end1) && (start2 != end2))
		{
			if(less(*start2, *start1))
			{
				merged_index.push_back(*start2);
				++ start2;
			}
			else	// start1 <= start2, start1 will be first
			{
				merged_index.push_back(*start1);
				++ start1;
			}
		}
		if(start1 != end1)
		{
			merged_index.insert(merged_index.end(), start1, end1);
		}
		else if(start2 != end2)
		{
			merged_index.insert(merged_index.end(), start2, end2);
		}
	}

	template<typename BufferSequence>
	std::size_t row_block::copy_data(const BufferSequence &data,
					 const std::size_t bytes)
	{
		typedef typename BufferSequence::const_iterator iter_type;
		
		iter_type iter = data.begin();
		const iter_type end = data.end();

		char * const storage = m_allocator.allocate(bytes);
		char * p = storage;
		for(; iter != end; ++ iter)
		{
			std::memcpy(p,
				    boost::asio::buffer_cast<const char*>(*iter),
				    boost::asio::buffer_size(*iter));

			const char * data_pos = p; // 新数据区开始位置
			p += boost::asio::buffer_size(*iter);
			const char * const data_end = p; // 新数据区结束位置

			// 添加记录的索引到index中
			while(data_pos < data_end)
			{
				const key_wrapper * const key = wrapper_cast<const key_wrapper*>(data_pos);
				data_pos += key->bytes();
				const value_wrapper * const value = wrapper_cast<const value_wrapper*>(data_pos);
				data_pos += value->bytes();

				// 总往后放，使索引的顺序与记录的顺序一致
				m_block_index.push_back(block_entry(key, value));
			}
			assert(data_pos == data_end);
		}
		assert(! m_block_index.empty());
		m_block = boost::asio::const_buffer(storage, bytes);
		m_row_name = wrapper_cast<const string_type*>(storage);

		// 合并索引，因为key相等Value不一定相等，所以新写入的数据必须要在前面
		std::stable_sort(m_block_index.begin(), m_block_index.end(),
				 row_block::entry_less_than()); // 先对新索引部分排序

		assert(std::size_t(p - storage) == bytes);
		return bytes;
	}

} // namespace xbase

#endif	// _XBASE_ROW_BLOCK_HPP_
